package Join;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ReduceJoinReducer extends Reducer<Text, ReduceJoinBean, ReduceJoinBean, NullWritable> {


    @Override
    protected void reduce(Text key, Iterable<ReduceJoinBean> values, Context context) throws IOException, InterruptedException {
        List<ReduceJoinBean> orderList = new ArrayList<ReduceJoinBean>();
        ReduceJoinBean pd = new ReduceJoinBean();
        for (ReduceJoinBean value : values) {
            if ("orders.txt".equals(value.getFileName())) {
                ReduceJoinBean reduceJoinBean = new ReduceJoinBean();
                try {
                    BeanUtils.copyProperties(reduceJoinBean,value);
                    //System.out.println(reduceJoinBean+reduceJoinBean.getFileName());
                    orderList.add(reduceJoinBean);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }else {
                try{
                    BeanUtils.copyProperties(pd,value);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
       for (ReduceJoinBean k:orderList){
           k.setPname(pd.getPname());
           k.setCategory_id(pd.getCategory_id());
           k.setPrice(pd.getPrice());
           System.out.println(k);
           context.write(k,NullWritable.get());
       }
    }
}
